package cn.stronglink.oil.mq;

import android.util.Log;

import com.google.gson.Gson;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.ExceptionHandler;
import com.rabbitmq.client.TopologyRecoveryException;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeoutException;

import cn.stronglink.oil.entity.DeviceAlarmInfo;
import cn.stronglink.oil.entity.DeviceRealTimeInfo;

/**
 * Created by tgb on 2017/4/18 18:14
 */

public class MqMessageReceiverClient {
    private final static int HISTORY_INFO_NUM = 300;
    private ConnectionFactory factory;
    private Channel channel;            // 实时数据管道
    private Channel alarmChannel;       // 警报管道
    private String exchangeName = "gas_collection";
    private String alarmExchangeName = "gas_alarm";        //
    private String routingKey;
    private String queueName;
    private String alarmQueueName;
    private IMqListener mqListener;
    private Connection realTimeConn;
    private Connection alarmConn;
    private Gson gson;
    private List<DeviceRealTimeInfo> deviceGasInfos;

    private boolean isConn;

    public IMqListener getMqListener() {
        return mqListener;
    }

    public void setMqListener(IMqListener mqListener) {
        this.mqListener = mqListener;
    }

    public MqMessageReceiverClient(String routingKey) {
        this.routingKey = routingKey;
        gson = new Gson();
        deviceGasInfos = new ArrayList<>();
        setConnectFactroy();
    }

    public List<DeviceRealTimeInfo> getDeviceGasInfos() {
        return deviceGasInfos;
    }

    private void addDeviceGasInfos(DeviceRealTimeInfo deviceGasInfo) {
        if (deviceGasInfos.size() >= HISTORY_INFO_NUM) {
            this.deviceGasInfos.remove(0);
        }
        deviceGasInfos.add(deviceGasInfo);
    }

    public void startConnect() {
        if (isConn)return;
        isConn = true;
        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    connect(routingKey);
                    alarmConect(routingKey);
                    while (true){
                        if (!isConn){
                            if (channel != null && channel.isOpen()) {
                                channel.close();
                            }
                            if (alarmChannel != null && alarmChannel.isOpen()) {
                                alarmChannel.close();
                            }

                            if (realTimeConn != null && realTimeConn.isOpen()) {
                                realTimeConn.close();
                            }
                            if (alarmChannel != null && alarmChannel.isOpen()) {
                                alarmChannel.close();
                            }
                            return;
                        }
                        Thread.sleep(1000);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                    if (getMqListener() != null) {
                        getMqListener().onDisConnect();
                    }
                }
            }
        }).start();
    }

    /**
     * 设置连接工厂
     */
    private void setConnectFactroy() {
        this.factory = new ConnectionFactory();
        this.factory.setUsername("robot");
        this.factory.setPassword("krundrobot");
        this.factory.setVirtualHost("shiyou");
        this.factory.setHost("www.yuzhantao.com");
        this.factory.setPort(5672);
        this.factory.setAutomaticRecoveryEnabled(true);
        this.factory.setRequestedHeartbeat(10);
        this.factory.setExceptionHandler(exceptionHandler);
    }

    /**
     * 连接通讯
     *
     * @param routingKey
     * @throws Exception
     */
    private void connect(String routingKey) throws Exception {
        realTimeConn = factory.newConnection();
        this.channel = realTimeConn.createChannel();
        this.channel.exchangeDeclare(exchangeName, "topic", true);
        queueName = channel.queueDeclare().getQueue();
        this.channel.queueBind(queueName, exchangeName, routingKey);
        if (this.getMqListener() != null) {
            this.getMqListener().onConnect();
        }
        this.consume();
    }

    private void alarmConect(String routingKey) throws Exception {
        alarmConn = factory.newConnection();
        this.alarmChannel = alarmConn.createChannel();
        this.alarmChannel.exchangeDeclare(alarmExchangeName, "topic", true);
        alarmQueueName = alarmChannel.queueDeclare().getQueue();
        this.alarmChannel.queueBind(alarmQueueName, alarmExchangeName, routingKey);
        if (this.getMqListener() != null) {
            this.getMqListener().onConnect();
        }
        this.AlarmConsume();
    }

    /**
     * 报警数据接收
     */
    private void AlarmConsume() throws Exception {
        boolean autoAck = false;
        alarmChannel.basicConsume(alarmQueueName, autoAck, new DefaultConsumer(alarmChannel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                super.handleDelivery(consumerTag, envelope, properties, body);
                String result = new String(body, "utf-8");
                Log.i("result", "result==" + result);
                DeviceAlarmInfo alarmInfo = gson.fromJson(result, DeviceAlarmInfo.class);
                if (alarmInfo != null && getMqListener() != null) {
                    getMqListener().onRecAlarmData(alarmInfo);
                }
            }
        });
    }

    /**
     * 发送消息
     *
     * @param bytes
     */
    public void pulish(byte[] bytes) {
        if (channel != null) {
            try {
                channel.basicPublish(exchangeName, routingKey, null, bytes);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 接收实时消息
     *
     * @throws Exception
     */
    private void consume() throws Exception {
        boolean autoAck = false;
        channel.basicConsume(queueName, autoAck, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                super.handleDelivery(consumerTag, envelope, properties, body);
                String result = new String(body, "utf-8");
                Log.i("result", "result==" + result);
                DeviceRealTimeInfo realTimeInfo = gson.fromJson(result, DeviceRealTimeInfo.class);
                if (realTimeInfo != null && getMqListener() != null) {
                    addDeviceGasInfos(realTimeInfo);
                    getMqListener().onRecData(realTimeInfo);
                }
            }
        });
    }

    /**
     * 异常处理
     */
    private ExceptionHandler exceptionHandler = new ExceptionHandler() {
        @Override
        public void handleUnexpectedConnectionDriverException(Connection conn, Throwable exception) {
            Log.i("mq", "handleUnexpectedConnectionDriverException");
            if (getMqListener() != null) {
                getMqListener().onDisConnect();
            }
        }

        @Override
        public void handleReturnListenerException(Channel channel, Throwable exception) {
            Log.i("mq", "handleReturnListenerException");
        }

        @Override
        public void handleFlowListenerException(Channel channel, Throwable exception) {
            Log.i("mq", "handleFlowListenerException");
        }

        @Override
        public void handleConfirmListenerException(Channel channel, Throwable exception) {
            Log.i("mq", "handleConfirmListenerException");
        }

        @Override
        public void handleBlockedListenerException(Connection connection, Throwable exception) {
            Log.i("mq", "handleBlockedListenerException");
        }

        @Override
        public void handleConsumerException(Channel channel, Throwable exception, Consumer consumer, String consumerTag, String methodName) {
            Log.i("mq", "handleConsumerException");
        }

        @Override
        public void handleConnectionRecoveryException(Connection conn, Throwable exception) {
            Log.i("mq", "handleConnectionRecoveryException");
        }

        @Override
        public void handleChannelRecoveryException(Channel ch, Throwable exception) {
            Log.i("mq", "handleChannelRecoveryException");
        }

        @Override
        public void handleTopologyRecoveryException(Connection conn, Channel ch, TopologyRecoveryException exception) {
            Log.i("mq", "handleTopologyRecoveryException");
        }
    };

    public void close() throws IOException, TimeoutException {
        this.isConn = false;
    }
}
